package org.piggy.core.filter.flowlimit;

import com.alibaba.csp.sentinel.SphO;
import com.alibaba.csp.sentinel.Tracer;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpResponse;
import org.piggy.common.constants.FilterConst;
import org.piggy.common.constants.OtherConst;
import org.piggy.common.enums.ResponseCode;
import org.piggy.common.exception.LimitedException;
import org.piggy.common.rule.Rule;
import org.piggy.core.context.Context;
import org.piggy.core.context.GatewayContext;
import org.piggy.core.filter.Filter;
import org.piggy.core.filter.FilterAspect;
import org.piggy.core.utils.ResponseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.function.Consumer;

@FilterAspect(id= FilterConst.FLOW_CTL_FILTER_ID, name = FilterConst.FLOW_CTL_FILTER_NAME, order = FilterConst.FLOW_CTL_FILTER_ORDER)
public class FlowLimitFilter implements Filter {

    private static final Logger accessLog = LoggerFactory.getLogger("accessLog");

    @Override
    public void doFilter(GatewayContext gatewayContext) {
        Rule.FlowControlConfig flowControlConfig = gatewayContext.getRule().getFlowControlConfig();
        if(flowControlConfig != null){
            Integer count = flowControlConfig.getCount();
            if(count != null && count > 0){
                Rule rule = gatewayContext.getRule();
                if(rule != null){
                    if(SphO.entry(OtherConst.FLOW + rule.getId())){
                        SphO.exit();
                    }else{
                        gatewayContext.terminated();
                        Consumer<Context> consumer = context -> {
                            context.getGatewayRequest().getFullHttpRequest().release();
                            ChannelHandlerContext channelHandlerContext = context.getChannelHandlerContext();
                            FullHttpResponse httpResponse = ResponseUtil.getHttpResponse(ResponseCode.FLOW_CONTROL_ERROR);
                            channelHandlerContext.writeAndFlush(httpResponse).addListener(ChannelFutureListener.CLOSE);
                            LimitedException limitedException = new LimitedException(ResponseCode.FLOW_CONTROL_ERROR);
                            Tracer.trace(limitedException);
                            accessLog.error("limit -> ", limitedException);
                        };
                        gatewayContext.setTerminatedCallBack(consumer);
                    }
                }
            }
        }
    }
}
